package com.study.flink.java.day04_async.async_io.es;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.elasticsearch.action.ActionFuture;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.transport.client.PreBuiltTransportClient;

import java.net.InetAddress;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.function.Supplier;

public class AsyncEsRequest extends RichAsyncFunction<String, Tuple2<String, String>> {

    private static transient TransportClient transportClient;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        Settings settings = Settings.builder().put("cluster.name", "HNS-ES").build();
        transportClient = new PreBuiltTransportClient(settings).addTransportAddresses(
            new InetSocketTransportAddress(InetAddress.getByName("192.168.50.31"), 9300),
            new InetSocketTransportAddress(InetAddress.getByName("192.168.50.31"), 9301),
            new InetSocketTransportAddress(InetAddress.getByName("192.168.50.34"), 9305),
            new InetSocketTransportAddress(InetAddress.getByName("192.168.50.34"), 9306)
        );

    }

    @Override
    public void close() throws Exception {
        super.close();
        transportClient.close();
    }

    @Override
    public void asyncInvoke(String key, ResultFuture<Tuple2<String, String>> resultFuture) throws Exception {
        ActionFuture<GetResponse> actionFuture = transportClient.get(new GetRequest("text_index", "users", key));
        CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                try {
                    GetResponse response = actionFuture.get();
                    return response.getSource().get("name").toString();
                } catch (InterruptedException | ExecutionException e) {
                    return null;
                }
            }
        }).thenAccept((String dbResult) -> {
            resultFuture.complete(Collections.singleton(new Tuple2<>(key, dbResult)));
        });
    }




}
